# -*- coding: UTF-8 -*-
from pyspark import SparkContext


# ********** Begin **********#
def f(iterator):
    list = []
    for x in iterator:
        list.append((x, len(x)))
    return list


# ********** End **********#

if __name__ == "__main__":
    # ********** Begin **********#

    # 1.初始化 SparkContext，该对象是 Spark 程序的入口
    sc = SparkContext('local', 'Simple App')

    # 2. 一个内容为（"dog", "salmon", "salmon", "rat", "elephant"）的列表List
    data = ["dog", "salmon", "salmon", "rat", "elephant"]

    # 3.通过 SparkContext 并行化创建 rdd
    rdd = sc.parallelize(data)

    # 4.使用rdd.collect() 收集 rdd 的元素。
    print(rdd.collect())

    """
    使用 mapPartitions 算子，将 rdd 的数据 ("dog", "salmon", "salmon", "rat", "elephant") 按照下面的规则进行转换操作，规则如下:
    需求：
        将字符串与该字符串的长度组合成一个元组，例如：
        dog  -->  (dog,3)
        salmon   -->  (salmon,6)
    """

    # 5.使用 mapPartitions 算子完成以上需求
    partitions = rdd.mapPartitions(f)

    # 6.使用rdd.collect() 收集完成 mapPartitions 转换的元素
    print(partitions.collect())

    # 7.停止 SparkContext
    sc.stop()

    # ********** End **********#

